package com.mini.task;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.mini.mysql.model.BusinessBaseInfo;
import com.mini.mysql.model.BusinessTeacherInfo;
import com.mini.es.model.EsBusinessStudentOrderInfo;
import com.mini.es.service.EsBusinessStudentOrderInfoService;
import com.mini.init.RobotInitRunnerImpl;
import com.mini.mysql.service.BaseService;
import com.mini.mysql.service.BusinessBaseInfoService;
import com.mini.mysql.service.BusinessStudentInfoService;
import com.mini.mysql.service.BusinessTeacherInfoService;
import com.mini.vo.BusinessStudentInfoVO;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.common.StopWatch;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigInteger;
import java.text.DecimalFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

@Service
@Slf4j
public class TaskAsyncMiniOrderDataServiceImpl {

    @Autowired
    private BusinessStudentInfoService businessStudentInfoService;
    @Autowired
    private EsBusinessStudentOrderInfoService esBusinessStudentOrderInfoService;
    @Autowired
    private BusinessTeacherInfoService businessTeacherInfoService;
    @Autowired
    private BusinessBaseInfoService businessBaseInfoService;
    @Autowired
    private RobotInitRunnerImpl robotInitRunner;

    @Resource(name = "asyncDataTaskExecutor")
    private ThreadPoolTaskExecutor asyncDataTaskExecutor;
    @Resource(name = "asyncData1TaskExecutor")
    private ThreadPoolTaskExecutor asyncData1TaskExecutor;
    @Resource(name = "asyncDataAttachTaskExecutor")
    private ThreadPoolTaskExecutor asyncDataAttachTaskExecutor;
    @Resource(name = "asyncDataTeacherTaskExecutor")
    private ThreadPoolTaskExecutor asyncDataTeacherTaskExecutor;
    @Resource(name = "asyncDataBaseInfoTaskExecutor")
    private ThreadPoolTaskExecutor asyncDataBaseInfoTaskExecutor;
    @Resource(name = "asyncDataClassTaskExecutor")
    private ThreadPoolTaskExecutor asyncDataClassTaskExecutor;
    @Resource(name = "asyncQueryDelDataTaskExecutor")
    private ThreadPoolTaskExecutor asyncQueryDelDataTaskExecutor;

    public static final DecimalFormat df = new DecimalFormat("0.00");
    @Value("${spring.profiles.active}")
    private String profiles;
    @Autowired
    private BaseService baseService;

    public static final Date initTime = DateUtil.parse("2000-01-01", DatePattern.NORM_DATE_PATTERN);

    /**
     * 根据updateTime传输数据 order表数据(使用线程池 asyncDataTaskExecutor)
     */
    @Scheduled(cron = "0/17 * * * * ?")
    public void asyncData() {
        if (profiles.contains("local") || profiles.contains("dev")) {
            return;
        }
        Future<?> future = asyncDataTaskExecutor.submit(() -> extractedOrder(null));
        monitor(future);
    }

    private void monitor(Future<?> future) {
        try {
            future.get(2, TimeUnit.HOURS);
        } catch (InterruptedException | TimeoutException | ExecutionException e) {
            log.error("", e);
            baseService.sendEmail("同步数据任务执行出错 ，请检查程序问题！");
        }
    }

    /**
     * 每天凌晨4点运行一次全量更新
     */
    @Scheduled(cron = "0 0 4 * * ?")
    public void asyncData1() {
        if (profiles.contains("local") || profiles.contains("dev")) {
            return;
        }
        Future<?> future = asyncData1TaskExecutor.submit(() -> extractedOrder(initTime));
        monitor(future);
    }


    /**
     * 根据updateTime传输数据 attach表数据（使用线程池 asyncDataAttachTaskExecutor）
     */
    @Scheduled(cron = "0/19 * * * * ?")
    public void asyncAttachData() {
        if (profiles.contains("local") || profiles.contains("dev")) {
            return;
        }
        Future<?> future = asyncDataAttachTaskExecutor.submit(() -> extractedOrderAttach(null));
        monitor(future);
    }

    /**
     * 根据updateTime传输数据 teacher表数据（使用线程池 asyncDataAttachTaskExecutor）
     */
    @Scheduled(cron = "0/15 * * * * ?")
    public void asyncOrderClassData() {
        if (
                profiles.contains("local") ||
                        profiles.contains("dev")) {
            return;
        }
        Future<?> future = asyncDataClassTaskExecutor.submit(() -> extractedOrderClass(null));
        monitor(future);
    }


    /**
     * 根据updateTime传输数据 class表数据（使用线程池 asyncDataAttachTaskExecutor）
     */
    @Scheduled(cron = "0/15 * * * * ?")
    public void asyncTeacherData() {
        if (profiles.contains("local") || profiles.contains("dev")) {
            return;
        }
        Future<?> future = asyncDataTeacherTaskExecutor.submit(() -> extractedOrderTeacher(null));
        monitor(future);
    }

    /**
     * 根据updateTime传输数据 teacher表数据（使用线程池 asyncDataAttachTaskExecutor）
     */
    @Scheduled(cron = "0/15 * * * * ?")
    public void asyncOrderBaseInfoData() {
        if (
                profiles.contains("local") ||
                        profiles.contains("dev")) {
            return;
        }
        Future<?> future = asyncDataBaseInfoTaskExecutor.submit(() -> extractedOrderBase(null));
        monitor(future);
    }

    /**
     * 检查最近7天内有没有删除的数据，如果有，一起删除ES中的数据
     */
    @Scheduled(cron = "0/60 * * * * ?")
    public void checkAndDelData() {
        if (
                profiles.contains("local") ||
                        profiles.contains("dev")) {
            return;
        }
        asyncQueryDelDataTaskExecutor.submit(() -> {
            try {
                checkAndDelData(null);
            } catch (Exception e) {
                log.info("", e);
            }
        });
    }

    /**
     * 刷新运营专员缓存信息
     */
    @Scheduled(cron = "0 0/5 * * * ?")
    public void refreshAllTeachers() {
        robotInitRunner.refreshAllTeachersAndGroups();
    }


    /**
     * ---------------------------------------非定时任务 分割线-----------------------------------------------
     */


    Date getDelay1Hour(Date startTime) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(startTime);
        calendar.set(Calendar.HOUR,
                calendar.get(Calendar.HOUR) + 1);
        return calendar.getTime();
    }


    public void checkAndDelData(Date startTime) throws InterruptedException {

        if (Objects.isNull(startTime)) {
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(new Date());
            calendar.set(Calendar.DAY_OF_YEAR,
                    calendar.get(Calendar.DAY_OF_YEAR) - 7);
            startTime = calendar.getTime();
        }

        Calendar calendar = Calendar.getInstance();
        calendar.setTime(new Date());
        calendar.set(Calendar.SECOND,
                calendar.get(Calendar.SECOND) - 60);
        Date now = calendar.getTime();

        Integer esCount = esBusinessStudentOrderInfoService.countByOrderPayTimeBetween(startTime, now);
        Integer mysqlCount = businessStudentInfoService.countByOrderPayTimeBetween(startTime, now);

        if (esCount > mysqlCount) {
            log.info("检查数据条数是否相同结果-不相同，开始删除ES中多余数据 esCount = {} mysqlCount = {}", esCount, mysqlCount);
            while (true) {
                Thread.sleep(10);
                Date endTime = getDelay1Hour(startTime);
                if (endTime.compareTo(now) > 0) {
                    endTime = now;
                }

                Integer thisEsCount = esBusinessStudentOrderInfoService.countByOrderPayTimeBetween(startTime, endTime);
                Integer thisMysqlCount = businessStudentInfoService.countByOrderPayTimeBetween(startTime, endTime);
                if (thisEsCount > thisMysqlCount) {
                    int i = 0;
                    PageRequest pageable = PageRequest.of(i, 10000);
                    Page<EsBusinessStudentOrderInfo> page = esBusinessStudentOrderInfoService.page(startTime, endTime, pageable);
                    List<EsBusinessStudentOrderInfo> esList = new ArrayList<>(page.getContent());

                    while (page.hasNext()) {
                        i++;
                        pageable = PageRequest.of(i, 10000);
                        page = esBusinessStudentOrderInfoService.page(startTime, endTime, pageable);
                        esList.addAll(page.getContent());
                    }

                    List<String> mysqlOrderNoList = businessStudentInfoService.findOrderNo(startTime, endTime);

                    esList = esList.stream().filter(order -> !mysqlOrderNoList.contains(order.getOrderNo())).collect(Collectors.toList());
                    if (!esList.isEmpty()) {
                        log.info("检查数据条数是否相同-分段删除-需要删除的数据 size = {} startTime = {} endTime= {}", esList.size(), startTime, endTime);
                        esBusinessStudentOrderInfoService.deleteAll(esList);
                        log.info("删除数据 {}", esList.stream().map(EsBusinessStudentOrderInfo::getOrderNo).collect(Collectors.toList()));
                    }
                }

                if (endTime.compareTo(now) >= 0) {
                    log.info("检查数据条数是否相同结果-共删除数据 {} 条任务结束！", esCount - mysqlCount);
                    return;
                }
                startTime = endTime;
            }

        } else {
            log.info("检查数据条数是否相同结果-无需处理！ esCount = {} mysqlCount = {} startTime = {} endTime = {} ", esCount, mysqlCount, startTime, now);
        }

    }


    public void extractedOrder(Date beginTime) {
        log.info("-----开始同步Order数据");
        //每次同步数据条数
        int pageSize = 10000;
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentOrderInfo order = esBusinessStudentOrderInfoService.findLatestUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getOrderUpdateTime())) {
            orderStartTime[0] = order.getOrderUpdateTime();
        }

        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        Page<BusinessStudentInfoVO> page = null;
        Integer total = businessStudentInfoService.countOrder(orderStartTime[0], now);
        int i = 0;
        try {
            long startTime = System.currentTimeMillis();
            final int[] pageNum = {0};
            while (i == 0 || page.hasNext()) {
                Pageable pageable = PageRequest.of(pageNum[0], pageSize, Sort.Direction.ASC, "update_time");
                i++;
                //mysql
                page = businessStudentInfoService.page(orderStartTime[0], now, pageable);
                int size = Math.min(pageSize * i, total);
                List<EsBusinessStudentOrderInfo> newList = page.getContent().stream().map(EsBusinessStudentOrderInfo::from).collect(Collectors.toList());
                if (newList.isEmpty()) {
                    log.info("-----任务结束-同步数据 条数 c = {} 进度： {}%", size, df.format((float) size / total * 100));
                    return;
                }

                //如果updateTime字段改变，则修改updateTime请求第1页数据，如果updateTime字段不变，则页码递增，请求第二页数据（优化查询速度）
                newList.stream().filter(o -> Objects.nonNull(o.getOrderUpdateTime())).max(Comparator.comparing(EsBusinessStudentOrderInfo::getOrderUpdateTime))
                        .ifPresent(o -> {
                            if (o.getOrderUpdateTime().compareTo(orderStartTime[0]) == 0) {
                                pageNum[0]++;
                            } else {
                                orderStartTime[0] = o.getOrderUpdateTime();
                                pageNum[0] = 0;
                            }
                        });

                //ES exist data
                List<String> orderNoList = newList.stream().map(EsBusinessStudentOrderInfo::getOrderNo).collect(Collectors.toList());
                List<EsBusinessStudentOrderInfo> existList = esBusinessStudentOrderInfoService.findAllById(orderNoList);

                //如果是已经存在的数据，不修改attachUpdateTime字段，防止数据互相覆盖造成误差
                newList = newList.stream().filter(newOrder -> {
                    EsBusinessStudentOrderInfo existOrder = existList.stream().filter(exist -> newOrder.getOrderNo().equals(exist.getOrderNo())).findFirst().orElse(null);
                    //es中不存在
                    if (Objects.isNull(existOrder)) {
                        return true;
                    }
                    //es中与mysql中数据相同
                    if (equal(newOrder, existOrder)) {
                        return false;
                    }
                    //数据不同
                    newOrder.setAttachUpdateTime(existOrder.getAttachUpdateTime());
                    newOrder.setTeacherUpdateTime(existOrder.getAttachUpdateTime());
                    newOrder.setClassUpdateTime(existOrder.getAttachUpdateTime());
                    return true;
                }).collect(Collectors.toList());

                //保存数据
                esBusinessStudentOrderInfoService.save(newList);

                long endTime = System.currentTimeMillis();
                float completion = (float) size / total;
                long totalSecond = (long) ((endTime - startTime) / 1000 / completion);
                if (size >= total) {
                    return;
                }
                log.info("本次同步数据 总条数 = {} 本次处理条数 c = {} 进度： {}% 需要总时长 {}分钟, 预计还需要{}分钟", total, size, df.format(completion * 100), df.format(totalSecond / 60), df.format((float) totalSecond / 60 * (1 - completion)));
            }
        } catch (Exception e) {
            log.error("", e);
        }

        log.info("结束同步Order数据，同步数据条数 {}", total);
    }

    private boolean equal(EsBusinessStudentOrderInfo newOrder, EsBusinessStudentOrderInfo existOrder) {
        return equal(newOrder.getOrderUpdateTime(), existOrder.getOrderUpdateTime())
                && equal(newOrder.getAttachUpdateTime(), existOrder.getAttachUpdateTime())
                && equal(newOrder.getTeacherUpdateTime(), existOrder.getTeacherUpdateTime())
                && equal(newOrder.getClassUpdateTime(), existOrder.getClassUpdateTime());
    }

    public static boolean equal(Date one, Date two) {
        return false;
//        if (Objects.isNull(one)) {
//            if (Objects.isNull(two)) {
//                return true;
//            }
//            return false;
//        } else {
//            if (Objects.isNull(two)) {
//                return false;
//            }
//            return one.compareTo(two) == 0;
//        }
    }


    public void extractedOrderAttach(Date beginTime) {
        log.info("-----开始同步attach数据");
        //每次同步数据条数
        int pageSize = 10000;

        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentOrderInfo order = esBusinessStudentOrderInfoService.findLatestAttachUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getAttachUpdateTime())) {
            orderStartTime[0] = order.getAttachUpdateTime();
        }

        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        Page<BusinessStudentInfoVO> page = null;
        Integer total = businessStudentInfoService.countAttach(orderStartTime[0], now);

        if (total <= 0) {
            log.info("需要同步数据为空，本次任务结束！");
            return;
        }
        int i = 0;
        try {
            long startTime = System.currentTimeMillis();
            final int[] pageNum = {0};
            while (i == 0 || page.hasNext()) {
                StopWatch stopWatch = new StopWatch();
                stopWatch.start("pageAttach");
                Pageable pageable = PageRequest.of(pageNum[0], pageSize);
                i++;
                page = businessStudentInfoService
                        .pageAttach(orderStartTime[0], now,
                                pageable);
                int size = Math.min(pageSize * i, total);
                List<EsBusinessStudentOrderInfo> newList = page.getContent().stream().map(EsBusinessStudentOrderInfo::from).collect(Collectors.toList());
                if (newList.isEmpty()) {
                    log.info("-----任务结束attach-同步数据 条数 c = {} 进度： {}%", size, df.format((float) size / total * 100));
                    return;
                }
                stopWatch.stop();

                //如果updateTime字段改变，则修改updateTime请求第1页数据，如果updateTime字段不变，则页码递增，请求第二页数据（优化查询速度）
                newList.stream().filter(o -> Objects.nonNull(o.getAttachUpdateTime())).max(Comparator.comparing(EsBusinessStudentOrderInfo::getAttachUpdateTime))
                        .ifPresent(o -> {
                            if (o.getAttachUpdateTime().compareTo(orderStartTime[0]) == 0) {
                                pageNum[0]++;
                            } else {
                                orderStartTime[0] = o.getAttachUpdateTime();
                                pageNum[0] = 0;
                            }
                        });
                stopWatch.start("ES findAllById");
                //ES exist data
                List<String> orderNoList = newList.stream().map(EsBusinessStudentOrderInfo::getOrderNo).collect(Collectors.toList());
                List<EsBusinessStudentOrderInfo> existList = esBusinessStudentOrderInfoService.findAllById(orderNoList);
                stopWatch.stop();
                stopWatch.start("edit");

                //如果是已经存在的数据，不修改attachUpdateTime字段，防止数据互相覆盖造成误差
                newList = newList.stream().filter(newOrder -> {
                    EsBusinessStudentOrderInfo existOrder = existList.stream().filter(exist -> newOrder.getOrderNo().equals(exist.getOrderNo())).findFirst().orElse(null);
                    //es中不存在
                    if (Objects.isNull(existOrder)) {
                        return true;
                    }
                    //es中与mysql中数据相同
                    if (equal(newOrder, existOrder)) {
                        return false;
                    }
                    newOrder.setOrderUpdateTime(existOrder.getOrderUpdateTime());
                    newOrder.setTeacherUpdateTime(existOrder.getTeacherUpdateTime());
                    newOrder.setClassUpdateTime(existOrder.getClassUpdateTime());
                    return true;
                }).collect(Collectors.toList());

                stopWatch.stop();
                stopWatch.start("save");
                esBusinessStudentOrderInfoService.save(newList);
                stopWatch.stop();

                long endTime = System.currentTimeMillis();
                float completion = (float) size / total;
                long totalSecond = (long) ((endTime - startTime) / 1000 / completion);
                log.info("本次同步数据-attach 总条数 = {} 本次处理条数 c = {} 进度： {}% 需要总时长 {}分钟, 预计还需要{}分钟", total, size, df.format(completion * 100), df.format(totalSecond / 60), df.format((float) totalSecond / 60 * (1 - completion)));
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }


    public void extractedOrderClass(Date beginTime) {
        log.info("-----开始同步class数据");
        //每次同步数据条数
        int pageSize = 10000;

        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentOrderInfo order = esBusinessStudentOrderInfoService.findLatestClassUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getClassUpdateTime())) {
            orderStartTime[0] = order.getClassUpdateTime();
        }

        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        Page<BusinessStudentInfoVO> page = null;
        Integer total = businessStudentInfoService.countClass(orderStartTime[0], now);

        if (total <= 0) {
            log.info("需要同步数据为空，本次任务结束！");
            return;
        }
        int i = 0;
        try {
            long startTime = System.currentTimeMillis();
            final int[] pageNum = {0};
            while (i == 0 || page.hasNext()) {
                StopWatch stopWatch = new StopWatch();
                stopWatch.start("pageClass");
                Pageable pageable = PageRequest.of(pageNum[0], pageSize);
                i++;
                //mysql
                page = businessStudentInfoService.pageClass(orderStartTime[0], now, pageable);
                int size = Math.min(pageSize * i, total);
                List<EsBusinessStudentOrderInfo> newList = page.getContent().stream().map(EsBusinessStudentOrderInfo::from).collect(Collectors.toList());
                if (newList.isEmpty()) {
                    log.info("-----任务结束class-同步数据 条数 c = {} 进度： {}%", size, df.format((float) size / total * 100));
                    return;
                }
                stopWatch.stop();

                stopWatch.start("ES findAllById");
                //ES exist data
                List<String> orderNoList = newList.stream().map(EsBusinessStudentOrderInfo::getOrderNo).collect(Collectors.toList());
                List<EsBusinessStudentOrderInfo> existList = esBusinessStudentOrderInfoService.findAllById(orderNoList);
                stopWatch.stop();
                stopWatch.start("edit");

                //如果updateTime字段改变，则修改updateTime请求第1页数据，如果updateTime字段不变，则页码递增，请求第二页数据（优化查询速度）
                newList.stream().filter(o -> Objects.nonNull(o.getClassUpdateTime())).max(Comparator.comparing(EsBusinessStudentOrderInfo::getClassUpdateTime))
                        .ifPresent(o -> {
                            if (o.getClassUpdateTime().compareTo(orderStartTime[0]) == 0) {
                                pageNum[0]++;
                            } else {
                                orderStartTime[0] = o.getClassUpdateTime();
                                pageNum[0] = 0;
                            }
                        });

                //如果是已经存在的数据，不修改attachUpdateTime字段，防止数据互相覆盖造成误差
                newList = newList.stream().filter(newOrder -> {
                    EsBusinessStudentOrderInfo existOrder = existList.stream().filter(exist -> newOrder.getOrderNo().equals(exist.getOrderNo())).findFirst().orElse(null);
                    //es中不存在
                    if (Objects.isNull(existOrder)) {
                        return true;
                    }
                    //es中与mysql中数据相同
                    if (equal(newOrder, existOrder)) {
                        return false;
                    }
                    newOrder.setOrderUpdateTime(existOrder.getOrderUpdateTime());
                    newOrder.setAttachUpdateTime(existOrder.getAttachUpdateTime());
                    newOrder.setTeacherUpdateTime(existOrder.getTeacherUpdateTime());
                    return true;
                }).collect(Collectors.toList());

                stopWatch.stop();
                stopWatch.start("save");
                esBusinessStudentOrderInfoService.save(newList);
                stopWatch.stop();

                long endTime = System.currentTimeMillis();
                float completion = (float) size / total;
                long totalSecond = (long) ((endTime - startTime) / 1000 / completion);
                log.info("本次同步数据-class 总条数 = {} 本次处理条数 c = {} 进度： {}% 需要总时长 {}分钟, 预计还需要{}分钟", total, size, df.format(completion * 100), df.format(totalSecond / 60), df.format((float) totalSecond / 60 * (1 - completion)));
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }

    public void extractedOrderTeacher(Date beginTime) {
        log.info("-----开始同步teacher数据");
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentOrderInfo order = esBusinessStudentOrderInfoService.findLatestTeacherUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getTeacherUpdateTime())) {
            orderStartTime[0] = order.getTeacherUpdateTime();
        }
        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        try {
            List<BusinessTeacherInfo> teachers = businessTeacherInfoService.findByUpdateTimeBetween(orderStartTime[0], now);
            updateByTeacher(teachers);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void updateByTeacher(List<BusinessTeacherInfo> teachers) {
        if (teachers.isEmpty()) {
            log.info("同步数据-teacher为空，任务结束！");
        }
        int total = 0;
        //每次同步数据条数
        int pageSize = 10000;
        Pageable pageable = PageRequest.of(0, pageSize);
        for (int i = 0; i < teachers.size(); i++) {
            BusinessTeacherInfo teacher = teachers.get(i);
            Page<EsBusinessStudentOrderInfo> existList = null;
            while (Objects.isNull(existList) || existList.hasNext()) {
                existList = esBusinessStudentOrderInfoService.pageByTeacherId((int) (long) teacher.getTeacherId(), teacher.getUpdateTime(), pageable);
                if (existList.getContent().isEmpty()) {
                    log.debug("teacher {} 没有查询到需要修改的数据，进行下一个任务", teacher.getTeacherId());
                    break;
                }
                existList.getContent().forEach(t -> {
                    t.setGroupId(BigInteger.valueOf(teacher.getGroupId()));
                    t.setTeacherUpdateTime(teacher.getUpdateTime());
                });
                esBusinessStudentOrderInfoService.save(existList.getContent());
                total += existList.getContent().size();
                log.info("本次同步数据-teacher = {}, 总条数 = {} 本次处理条数 {}, 已经修改teacher size = {} 一共需要修改的teacher size = {}", teacher.getTeacherId(), existList.getTotalElements(), existList.getContent().size(), i + 1, teachers.size());

            }
        }
        log.info("本次同步数据-teacher结束! 同步数据 total = {}", total);
    }

    public void extractedOrderBase(Date beginTime) {
        log.info("-----开始同步base_info数据");
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentOrderInfo order = esBusinessStudentOrderInfoService.findLatestBaseUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getBaseUpdateTime())) {
            orderStartTime[0] = order.getBaseUpdateTime();
        }
        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        try {
            List<BusinessBaseInfo> baseInfos = businessBaseInfoService.findByUpdateTimeBetween(orderStartTime[0], now);
            updateByBaseInfo(baseInfos);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void updateByBaseInfo(List<BusinessBaseInfo> baseInfos) {
        if (baseInfos.isEmpty()) {
            log.info("同步数据-base为空，任务结束！");
        }
        int total = 0;
        //每次同步数据条数
        int pageSize = 10000;
        Pageable pageable = PageRequest.of(0, pageSize);
        for (int i = 0; i < baseInfos.size(); i++) {
            BusinessBaseInfo baseInfo = baseInfos.get(i);
            Page<EsBusinessStudentOrderInfo> existList = null;
            while (Objects.isNull(existList) || existList.hasNext()) {
                existList = esBusinessStudentOrderInfoService.pageByMiniLessonId(baseInfo.getMiniLessonId(), baseInfo.getUpdateTime(), pageable);
                if (existList.getContent().isEmpty()) {
                    log.debug("baseInfo {} 没有查询到需要修改的数据，进行下一个任务", baseInfo.getMiniLessonId());
                    break;
                }
                existList.getContent().forEach(t -> {
                    t.setWorkRoomId(BigInteger.valueOf(baseInfo.getWorkRoomId()));
                    t.setWorkRoomName(baseInfo.getWorkRoomName());
                    t.setChannelCode(baseInfo.getChannelCode());
                    t.setChannelName(baseInfo.getChannelName());
                    t.setBaseUpdateTime(baseInfo.getUpdateTime());
                });
                esBusinessStudentOrderInfoService.save(existList.getContent());
                total += existList.getContent().size();
//                log.info("本次同步数据-baseInfo = {}, 总条数 = {} 本次处理条数 {}, 已经修改baseInfo size = {} 一共需要修改的teacher size = {}", baseInfo.getMiniLessonId(), existList.getTotalElements(), existList.getContent().size(), i + 1, baseInfos.size());

            }
        }
        log.info("本次同步数据-baseInfo结束! 同步数据 total = {}", total);
    }

}